use std::{
    collections::{HashSet, LinkedList},
    sync::{atomic::AtomicI64, Arc, Mutex},
};

use async_trait::async_trait;
use common::{
    base::id_generator,
    protocol::{get_sender_service_key, protocol_v1::ProtocolV1Writer, MsgType, XID},
    serial::api_descriptor::ServiceApiDescriptor,
    service::{sys_service_api::xport_service::ServiceInfo, ServiceKey},
    utils::time_utils,
};
use semver::Version;
use tokio::sync::mpsc::{self, UnboundedReceiver, UnboundedSender};
use tracing::{debug, info, warn};

use crate::{airport::airport::Airport, log, protocol::MsgPriority};

use super::{loader::ServiceIns, service::Service};

pub struct LocalService {
    dxc_name: String,
    /**
     *
     */
    pub airport: Arc<Box<Airport>>,

    pub api_descriptor: ServiceApiDescriptor,

    /**
     *
     */
    dxc_version: Version,
    /**
     *
     */
    service_name: String,

    service_id: i64,

    md5: String,
    /**
     *
     */
    is_system: bool,

    /**
     * 上线时间
     */
    online_time: AtomicI64,
    /**
     * 主节点id
     */
    master_id: i64,

    /**
     *  加载的 service 实例
     */
    service_ins: Box<dyn ServiceIns>,
    /**
     * 通知有消息
     */
    message_receiver: tokio::sync::Mutex<UnboundedReceiver<bool>>,

    /**
     * 消息借
     */
    message_sender: UnboundedSender<bool>,
    //
    reponse_message_receiver: tokio::sync::Mutex<UnboundedReceiver<bool>>,

    /**
     *
     */
    reponse_message_sender: UnboundedSender<bool>,

    low_pri_message_list: Mutex<LinkedList<XID>>,
    /**
     * 低优先级的消息
     */
    normal_pri_message_list: Mutex<LinkedList<XID>>,
    /**
     * 低优先级的消息
     */
    high_pri_message_list: Mutex<LinkedList<XID>>,
    // 等待回复的信息
    // waitting_request_id_set: Mutex<HashSet<i64>>,
}

impl LocalService {
    pub fn new(
        dxc_name: String,
        dxc_version: Version,
        dxc_id: i64,
        md5: String,
        is_system: bool,
        service_name: String,
        airport: Arc<Box<Airport>>,
        service_ins: Box<dyn ServiceIns>,
        api_descriptor: ServiceApiDescriptor,
    ) -> Box<LocalService> {
        let (sender, receiver) = mpsc::unbounded_channel::<bool>();
        let (resp_sender, resp_receiver) = mpsc::unbounded_channel::<bool>();

        Box::new(LocalService {
            dxc_name,
            service_id: id_generator::gen_service_id(dxc_id),
            airport,
            master_id: 0,
            md5,
            online_time: AtomicI64::new(0),
            dxc_version: dxc_version,
            is_system: is_system,
            service_name,
            service_ins,
            message_receiver: tokio::sync::Mutex::new(receiver),
            message_sender: sender,

            reponse_message_receiver: tokio::sync::Mutex::new(resp_receiver),
            reponse_message_sender: resp_sender,

            api_descriptor,
            low_pri_message_list: Mutex::new(LinkedList::new()),
            normal_pri_message_list: Mutex::new(LinkedList::new()),
            high_pri_message_list: Mutex::new(LinkedList::new()),
        })
    }

    pub fn dxc_name(&self) -> String {
        self.dxc_name.clone()
    }

    pub fn md5(&self) -> String {
        self.md5.clone()
    }

    fn get_xid(&self) -> Option<XID> {
        let xid = {
            let mut message_list = self.high_pri_message_list.lock().unwrap();

            message_list.pop_front()
        };

        if xid.is_some() {
            return xid;
        }

        let xid = {
            let mut message_list = self.low_pri_message_list.lock().unwrap();
            message_list.pop_front()
        };

        xid
    }

    pub fn is_system(&self) -> bool {
        self.is_system
    }

    async fn handle_message(self: Arc<Box<Self>>) {
        let mut message_receiver = self.message_receiver.lock().await;

        while let Some(is_handle) = message_receiver.recv().await {
            if !is_handle {
                break;
            }
            let xid = self.get_xid();
            // 处理消息
            if xid.is_none() {
                continue;
            }
            let xid = xid.unwrap();
            let message_manager = self.airport.get_message_manager();
            let message = message_manager.get_msg(&xid).await;
            if message.is_none() {
                continue;
            }
            let clone_self = self.clone();
            let self_service_key = self.get_service_key();

            tokio::spawn(async move {
                let (message, channel_id) = message.unwrap();
                let rsp_message = clone_self
                    .service_ins
                    .dispatch_message(channel_id, xid.request_id, &message)
                    .await;

                if rsp_message.is_err() {
                    warn!("{}", rsp_message.err().unwrap());
                    return;
                }
                //

                let rsp_message = rsp_message.unwrap();

                // 从里面读取

                let sender_service_key = get_sender_service_key(message.as_slice()).unwrap();

                let mut v1_writer = ProtocolV1Writer::new(
                    rsp_message.len() as u32,
                    xid.request_id,
                    &self_service_key,
                    &sender_service_key,
                );

                v1_writer.write_msg_body(&rsp_message);
                v1_writer.write_msg_type(MsgType::RspNormalMsg);

                clone_self
                    .airport
                    .response_message(xid, v1_writer.msg_buffer)
                    .await;

                message_manager.delete_message(xid).await;
            });

            //
        }

        message_receiver.close();

        //
    }

    async fn handle_response_message(self: Arc<Box<Self>>) {
        let mut reponse_message_receiver = self.reponse_message_receiver.lock().await;

        while let Some(is_handle) = reponse_message_receiver.recv().await {
            if !is_handle {
                break;
            }

            let xid = {
                let mut message_list = self.normal_pri_message_list.lock().unwrap();

                message_list.pop_front()
            };
            // 处理消息
            if xid.is_none() {
                continue;
            }
            let xid = xid.unwrap();

            debug!("handle_response_message 返回消息：{}", xid.request_id);

            let message_manager = self.airport.get_message_manager();

            let message = message_manager.get_msg(&xid).await;

            if message.is_none() {
                continue;
            }
            let clone_self = self.clone();

            tokio::spawn(async move {
                let (message, _) = message.unwrap();

                clone_self.service_ins.result_in(xid.request_id, &message);

                message_manager.delete_message(xid).await;
            });
        }

        reponse_message_receiver.close();

        //
    }

    pub fn start(self: &Arc<Box<Self>>) {
        self.online_time.store(
            time_utils::cur_timestamp(),
            std::sync::atomic::Ordering::Relaxed,
        );

        let clone_self = self.clone();

        tokio::spawn(async move {
            clone_self.handle_message().await;
        });

        let clone_self = self.clone();
        tokio::spawn(async move {
            clone_self.handle_response_message().await;
        });
    }

    pub fn get_service_info(&self) -> ServiceInfo {
        let mut local_service_info = ServiceInfo::default();

        local_service_info.dxc_name = self.dxc_name();
        local_service_info.dxc_version = self.dxc_version.to_string();
        local_service_info.service_name = String::from(self.name());
        local_service_info.service_id = self.service_id;
        local_service_info.md5 = self.md5();
        local_service_info.online_time = self.oline_time();
        local_service_info
    }
}

impl Drop for LocalService {
    fn drop(&mut self) {
        info!("卸载local_service :{:?}", self.get_service_key());
    }
}

#[async_trait]
impl Service for LocalService {
    fn id(&self) -> i64 {
        self.service_id
    }

    fn version(&self) -> Version {
        self.dxc_version.clone()
    }

    fn name(&self) -> &str {
        &self.service_name
    }

    fn init(&self, config: &str) {
        let log_level = log::level();

        self.service_ins.init(self.service_id, config, log_level)
    }

    async fn stop(&self) {
        //
        let _ = self.message_sender.send(false);
        let _ = self.reponse_message_sender.send(false);
        self.message_sender.closed().await;
    }

    fn dispatch_message_with_pri(&self, pri: MsgPriority, xid: XID) {
        match pri {
            MsgPriority::Low => {
                let mut msg_list = self.low_pri_message_list.lock().unwrap();
                msg_list.push_back(xid)
            }
            MsgPriority::Normal => {
                let mut msg_list = self.normal_pri_message_list.lock().unwrap();
                msg_list.push_back(xid)
            }
            MsgPriority::Hig => {
                let mut msg_list = self.high_pri_message_list.lock().unwrap();

                msg_list.push_back(xid)
            }
        };

        let result = if pri == MsgPriority::Normal {
            self.reponse_message_sender.send(true)
        } else {
            self.message_sender.send(true)
        };

        if result.is_err() {
            warn!("service :{} 接收消息出错！", self.service_name)
        }
    }

    fn finalize(&self) {
        self.service_ins.finalize();
    }

    fn get_service_key(&self) -> ServiceKey {
        ServiceKey::new(
            self.dxc_name.clone(),
            self.dxc_version.to_string(),
            self.service_name.clone(),
        )
    }

    fn execute_parallel_task(&self, _task_id: i64) -> bool {
        false
    }

    fn disable_register(&self) {}

    fn is_register(&self) -> bool {
        true
    }

    fn set_private(&self, private: bool) {}

    fn is_public(&self) -> bool {
        false
    }

    fn set_master_id(&mut self, master_id: i64) {
        self.master_id = master_id;
    }

    fn master_id(&self) -> i64 {
        self.master_id
    }

    fn oline_time(&self) -> i64 {
        self.online_time.load(std::sync::atomic::Ordering::Relaxed)
    }

    fn set_token_id(&self, token_id: i64) {}

    fn token_id(&self) -> i64 {
        0
    }
}
